package com.seventeen.mq.proxy.server.proxy.ext;

import com.seventeen.mq.proxy.common.utils.StringUtils;
import com.seventeen.mq.proxy.server.proxy.MqClient;
import com.seventeen.mq.proxy.server.proxy.consumer.ConsumeStatus;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;

/**
 * 4.x 消费者
 *
 * @author seventeen
 */
public class NewExtMqPushConsumer extends BaseExtMqConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(NewExtMqPushConsumer.class);
    private PushConsumer pushConsumer;

    /**
     * 初始化
     */
    public void init() {
        try {
            ClientConfigurationBuilder builder = ClientConfiguration.newBuilder()
                    .setEndpoints(endpoints);
            if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey)) {
                SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
                builder.setCredentialProvider(sessionCredentialsProvider);
            }
            ClientConfiguration clientConfiguration = builder.build();
            FilterExpression filterExpression = new FilterExpression(tags, FilterExpressionType.TAG);
            // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
            ClientServiceProvider provider = ClientServiceProvider.loadService();
            PushConsumerBuilder pushConsumerBuilder = provider.newPushConsumerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    // 设置消费者分组。
                    .setConsumerGroup(consumerProxyGroup)
                    // 设置预绑定的订阅关系。
                    .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression));
            pushConsumer = pushConsumerBuilder
                    // 设置消费监听器。
                    .setMessageListener(messageView -> {
                        String convertJson = ExtProxyMessage.convert(messageView);
                        ConsumeStatus consumeStatus = MqClient.consumer(appUrl, convertJson);
                        if (consumeStatus == ConsumeStatus.SUCCESS) {
                            //根据消费结果返回状态。
                            return ConsumeResult.SUCCESS;
                        }
                        return ConsumeResult.FAILURE;
                    }).build();
            LOGGER.info("{} started!", this);
        } catch (Exception e) {
            LOGGER.error("NewExtMQProducer start fail:{},{}", consumerProxyGroup, getEndpoints(), e);
        }
    }

    public void destroy() {
        LOGGER.info("{} shutdown!", this);
        if (pushConsumer != null) {
            try {
                pushConsumer.close();
            } catch (Exception e) {
                LOGGER.error("{}  shutdown error,", this, e);
            }
        }
    }

    @Override
    public String toString() {
        return "NewExtMqPushConsumer{" +
                "consumer=" + pushConsumer +
                ", nameServerAddr='" + endpoints + '\'' +
                ", consumerProxyGroup='" + consumerProxyGroup + '\'' +
                ", batchMaxSize=" + batchMaxSize +
                ", topic='" + topic + '\'' +
                ", tags='" + tags + '\'' +
                ", appUrl='" + appUrl + '\'' +
                ", type='" + type + '\'' +
                ", version='" + version + '\'' +
                '}';
    }

}
